package top.fullj.actor;

import top.fullj.actor.exception.InvalidOperationException;
import top.fullj.actor.mq.*;

import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.List;
import java.util.Stack;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Actor's context
 *  provide important method for Actor
 *
 * @author wurenhai
 * @since 2018/3/11
 */
public class ActorContext {

    /**
     * child Actors
     */
    private final List<ActorRef> children = new CopyOnWriteArrayList<>();

    /**
     * parent Actor
     */
    public final ActorRef parent;

    public final ActorPath path;

    /**
     * arguments, for recover
     */
    public final Props props;

    /**
     * self ActorRef
     */
    public final ActorRef self;

    /**
     * ActorSystem reference
     */
    public final ActorSystem system;

    /**
     * terminated or not
     */
    private volatile boolean terminated;

    public List<ActorRef> children() {
        return Collections.unmodifiableList(children);
    }

    public boolean isTerminated() {
        return terminated;
    }

    public ActorRef actorOf(Class<? extends Actor> type, Props props) {
        return actorOf(type, props, generateActorName());
    }

    /**
     * create child Actor
     */
    public ActorRef actorOf(Class<? extends Actor> type, Props props, String name) {
        validateName(name);
        Actor actor = ActorCreator.createWith(type, props);
        ActorContext context = new ActorContext(system,
                type,
                actor,
                name,
                props,
                this.self,
                this.path);
        children.add(context.self);
        return context.self;
    }

    /**
     * generate ActorSelection path, unsupported wildcard
     *
     * @param path path to select Actors
     */
    public ActorSelection select(String path) {
        if (path.startsWith("/")) {
            return system.select(path);
        }
        return select(false, path);
    }

    public void become(Receiver receiver) {
        become(receiver, true);
    }

    /**
     * change the message handler
     *
     * @param receiver message handler
     * @param discardOld discard preview or not
     */
    public void become(Receiver receiver, boolean discardOld) {
        if (discardOld) {
            unbecome();
        }
        this.receivers.push(receiver);
        channel.clearSubscribers();
        channel.subscribe(fiber, getOnReceiveDelegate(receiver));
    }

    /**
     * stop the referenced actor
     */
    public void stop(ActorRef actor) {
        if (actor == self) {
            parent.ctx.stop(self);
        } else {
            if (children.remove(actor)) {
                actor.ctx.stop();
            } else {
                sweep(actor, children());
            }
        }
    }

    /**
     * subscribe message with preview Receiver
     */
    @SuppressWarnings("SpellCheckingInspection")
    public void unbecome() {
        if (receivers.size() > 1) {
            receivers.pop();//pop current receiver first
        }
        Receiver receiver = receivers.pop();
        channel.clearSubscribers();
        channel.subscribe(fiber, getOnReceiveDelegate(receiver));
    }

    ActorContext(ActorSystem system,
                 Class<? extends Actor> actorType,
                 Actor actor,
                 String name,
                 Props props,
                 ActorRef parent,
                 ActorPath parentPath) {
        this(system, actorType, actor, name, props, parent, parentPath, false,
                new InMemoryChannel<>(), FiberPool.assign(name));
    }

    private ActorContext(ActorSystem system,
                         Class<? extends Actor> actorType,
                         Actor actor,
                         String name,
                         Props props,
                         ActorRef parent,
                         ActorPath parentPath,
                         boolean suspended,
                         Channel<Delivery> channel,
                         Fiber fiber) {
        this.system = system;
        this.actor = actor;
        this.actor.setContext(this);
        this.channel = channel;
        this.fiber = fiber;
        this.parent = parent;
        this.path = parentPath.withName(name);
        this.props = props;
        this.self = new ActorRef(this);
        this.suspended = suspended;
        this.terminated = false;
        this.type = actorType;

        start();
    }

    /**
     * type of the Actor instance, for recover
     */
    private final Class<? extends Actor> type;

    /**
     * my self Actor
     */
    private final Actor actor;

    /**
     * channel to pub/sub messages
     */
    private final Channel<Delivery> channel;

    /**
     * handle messages in fiber
     */
    private final Fiber fiber;

    /**
     * current suspended state
     */
    private volatile boolean suspended;

    Actor actor() {
        return actor;
    }

    void enqueue(Delivery delivery) {
        channel.publish(delivery);
    }

    boolean hasChildren() {
        return !children.isEmpty();
    }

    void stop() {
        suspended = true;
        actor.stopChildren();
        terminated = true;
        stopConcurrency();
        actor.postStop();
        stopContext();
    }

    /**
     * receiver stack
     */
    private final Stack<Receiver> receivers = new Stack<>();

    /**
     * used to name Actor without name
     */
    private final AtomicInteger nameIndex = new AtomicInteger(1);

    /**
     * create new Actor with this context
     */
    private ActorRef actorOf(Class<? extends Actor> actorType,
                             Props props,
                             String name,
                             boolean suspended,
                             Channel<Delivery> channel,
                             Fiber fiber) {
        validateName(name);
        Actor actor = ActorCreator.createWith(actorType, props);
        ActorContext context = new ActorContext(system,
                actorType,
                actor,
                name,
                props,
                self,
                path,
                suspended,
                channel,
                fiber);
        children.add(context.self);
        return context.self;
    }

    ActorSelection select(boolean root, String path) {
        // split root path will make the first element empty
        return select(root, path.split("/"), root ? 1 : 0);
    }

    private ActorSelection select(boolean root, String[] path, int index) {
        if (root && ActorPath.USR.equals(path[index])) {
            ++index;
        }

        //TODO currently only support ends(*) wildcard pattern(/usr/*) to select children
        if ((index + 1) == path.length
                && ActorPath.ANY.equals(path[index])) {
            return children.isEmpty()
                    ? new ActorSelection(system.deadLetters)//if no child
                    : new ActorSelection(children);
        }

        //select name full matched child
        for (ActorRef child : children) {
            if (child.path().name().equals(path[index])) {
                return ((index + 1) == path.length)
                    ? new ActorSelection(child)
                    : child.ctx.select(false, path, index + 1);
            }
        }
        return new ActorSelection(system.deadLetters);
    }

    private void checkSuspended() {
        while (suspended) {
            if (terminated) {
                break;
            }
            try {
                //noinspection BusyWait
                Thread.sleep(1);
            } catch (InterruptedException ignore) {
                break;
            }
        }
    }

    private String generateActorName() {
        int value = nameIndex.getAndIncrement();
        byte[] bytes = toBytes(value);
        return "$" + bytesToHexString(bytes);
    }

    private byte[] toBytes(int value) {
        ByteBuffer buf = ByteBuffer.allocate(Integer.BYTES);
        buf.putInt(value);
        return buf.array();
    }

    private static final byte[] TO_HEX = {
            '0', '1', '2', '3', '4', '5', '6', '7',
            '8', '9', 'a', 'b', 'c', 'd', 'e', 'f',
    };

    private String bytesToHexString(byte[] src) {
        byte[] out = new byte[src.length << 1];
        for (int i = 0; i < src.length; i++) {
            byte b = src[i];
            out[ i<<1 ] = TO_HEX[(b >> 4) & 0x0F];
            out[(i<<1) + 1] = TO_HEX[b & 0x0F];
        }
        return new String(out);
    }

    /**
     * delegate around call OnReceive method
     */
    private Action<Delivery> getOnReceiveDelegate(final Receiver receiver) {
        return delivery -> {
            checkSuspended();
            try {
                Method m = receiver.getClass().getDeclaredMethod("onReceive",
                        ActorRef.class, delivery.message.getClass());
                m.setAccessible(true);
                m.invoke(receiver, delivery.sender, delivery.message);
            } catch (NoSuchMethodException ignore) {
                receiver.onReceive(delivery.sender, delivery.message);
            } catch (Exception e) {
                handleFailure(e, delivery.message);
            }
        };
    }

    /**
     * if handle message failed, let its parent decide what to do
     */
    private void handleFailure(Exception reason, Object message) {
        parent.ctx.recoverFromFor(reason, message, self);
    }

    /**
     * Escalate the failure to my parent
     */
    private void recoverByEscalation(Exception reason) {
        parent.ctx.recoverFromFor(reason, null, self);
    }

    private void recoverByRestarting(Exception reason, Object message, ActorRef child) {
        ActorContext childContext = child.ctx;
        try {
            childContext.suspended = true;
            childContext.actor.preRestart(reason, message);
            children.remove(child);
            ActorRef restarted = actorOf(childContext.type,
                    childContext.props,
                    childContext.path.name(),
                    true,
                    childContext.channel,
                    childContext.fiber);
            restarted.ctx.suspended = false;
            childContext.stopContext();//release old child context
            restarted.ctx.actor().postRestart(reason);
        } finally {
            childContext.suspended = false;
        }
    }

    @SuppressWarnings("unused")
    private void recoverByResume(Exception reason, ActorRef child) {
        //ignore
    }

    /**
     * stop my failed child
     */
    @SuppressWarnings("unused")
    private void recoverByStopping(Exception reason, ActorRef child) {
        stop(child);
    }

    private void recoverFromFor(Exception reason, Object message, ActorRef child) {
        Actor.SupervisorStrategy.Directive directive = actor.strategy.decide(reason);
        switch (directive) {
            case Restart:
                recoverByRestarting(reason, message, child);
                break;
            case Stop:
                recoverByStopping(reason, child);
                break;
            case Escalate:
                recoverByEscalation(reason);
                break;
            case Resume:
                recoverByResume(reason, child);
                break;
        }
    }

    private void start() {
        become(actor, false);
        if (!suspended) {
            actor.preStart();
            fiber.start();
        }
    }

    private void stopConcurrency() {
        channel.clearSubscribers();
        fiber.dispose();
    }

    private void stopContext() {
        this.actor.setContext(null);
        this.children.clear();
        this.receivers.clear();
        this.suspended = true;
        this.terminated = true;
    }

    /**
     * stop sub Actor tree at any level
     */
    private void sweep(ActorRef actor, List<ActorRef> children) {
        for (ActorRef child : children) {
            if (child == actor) {
                child.ctx.parent.ctx.stop(child);
                return;
            }
        }
        for (ActorRef child : children) {
            sweep(actor, child.ctx.children());
        }
    }

    private void validateName(String name) {
        if (name == null || name.isEmpty()) {
            throw new InvalidOperationException("Actor name is required");
        }
        if (name.substring(1).contains("$") || name.contains("/")) {
            throw new InvalidOperationException("Invalid actor name: " + name);
        }
        for (ActorRef child : children()) {
            if (name.equals(child.path().name())) {
                throw new InvalidOperationException("Duplicated actor name: " + name);
            }
        }
    }

    static class Delivery {

        private final Object message;
        private final ActorRef sender;

        Delivery(Object message, ActorRef sender) {
            this.message = message;
            this.sender = sender;
        }

    }

    /**
     * message receiver
     */
    public interface Receiver {
        void onReceive(ActorRef sender, Object message);
    }

    static class ActorCreator {

        static Actor createWith(Class<?> type, Props props) {
            int count = props.getCount();
            int index = 0;
            Class<?>[] types = new Class<?>[count];
            Object[] args = new Object[count];
            for (Object arg : props) {
                types[index] = arg.getClass();
                args[index] = arg;
                ++index;
            }
            try {
                Constructor<?> constructor = type.getDeclaredConstructor(types);
                constructor.setAccessible(true);
                return (Actor)constructor.newInstance(args);
            } catch (ReflectiveOperationException e) {
                throw new InvalidOperationException("No constructor of type: " + type.getName());
            }
        }

    }

}
